package syR

import (
	"errors"
	"fmt"
	"db2s/Fp"
	"db2s/global"
	"db2s/go-log/log"
	"db2s/parDef"
	"db2s/ref"
	mq "db2s/topic-mq"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
)

func syncBaseMissQuery(wg *sync.WaitGroup, message mq.Message) (err error) {
	var event = "[syncBaseMissQuery]"
	defer func() {
		err = ref.RecoverPanic(event, recover())
	}()
	var plan = message.Properties["plan"].(*SchedulePlan2)
	switch {
	case len(plan.TableAttributes.WhereSql) > 0: //where sql
		plan.MQ.FirstObjectProductAdd()
		err = whereSqlFilter(wg, message)
	default:
		plan.MQ.FirstObjectProductSwap(int64(len(plan.Object.PartName)))
		err = missSendMessage(wg, message)
	}
	return
}
func syncBaseIndexQuery(wg *sync.WaitGroup, message mq.Message) (err error) {
	var event = "[syncBaseIndexQuery]"
	defer func() {
		if r := recover(); r != nil {
			err = recoverCost(event, r)
			log.ErrorLog().Error(fmt.Sprintf("%v", err))
		}
	}()
	var plan = message.Properties["plan"].(*SchedulePlan2)
	switch {
	case len(plan.TableAttributes.WhereSql) > 0: //where sql
		plan.MQ.FirstObjectProductAdd()
		err = whereSqlFilter(wg, message)
	default:
		plan.MQ.FirstObjectProductSwap(int64(len(plan.Object.PartName)))
		err = IsSegmentSendMessage(wg, plan.Object.ObjectLogSeq, message)
	}
	return
}
func syncBaseQuery(wg *sync.WaitGroup, message mq.Message) (err error) {
	var event = "[syncBaseQuery]"
	defer func() {
		if r := ref.RecoverPanic(event, recover()); err == nil && r != nil {
			err = r
		}
		if err != nil {
			handleError(message, event, err)
		}
	}()
	var plan = message.Properties["plan"].(*SchedulePlan2)
	switch plan.baseMe.Index.(type) {
	case parDef.IndexColumnMe:
		switch plan.baseMe.Index.(parDef.IndexColumnMe).ColumnKey {
		case "P", "U", "I":
			plan.MQ.SpeedLimitSwitch = true
			plan.MQ.Speed.SpeedMonitor()
			err = syncBaseIndexQuery(wg, message)
		default:
			plan.MQ.Speed.SpeedMonitor()
			err = syncBaseMissQuery(wg, message)
		}
	default:
		plan.MQ.Speed.SpeedMonitor()
		err = syncBaseMissQuery(wg, message)
	}
	return
}
func syncGlobalInitActive(message mq.Message) (err error) {
	var event = "[syncGlobalInitActive]"
	defer func() {
		if r := recover(); r != nil {
			err = recoverCost(event, r)
		}
		if err != nil {
			handleError(message, event, err)
		}
	}()
	var plan = message.Properties["plan"].(*SchedulePlan2)
	plan.Status.SubTaskBeginTime = time.Now().UnixNano()
	if GlobalPConfigs.result.Teletypewriter == "bar" {
		PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "running")
		log.MainLog().Info(fmt.Sprintf("(%d) %s Add task Bar Run object %v successfully", plan.Object.ObjectLogSeq, event, fmt.Sprintf("%v.%v", plan.Object.Schema, plan.Object.Table)))
	}
	//需要判断当前数据库是否有锁
	return truncateTableActive(plan.Object.PartName, plan)
}
func speedAdd(event string, plan1 any) (res any) {
	plan := plan1.(*SchedulePlan2)
	if atomic.LoadInt64(plan.ReadOptimizer.Thread) > 5 {
		atomic.SwapInt64(plan.ReadOptimizer.Thread, atomic.LoadInt64(plan.ReadOptimizer.Thread)-1)
	}
	if atomic.LoadInt64(plan.writeOptimizer.Thread) > 1 {
		atomic.SwapInt64(plan.writeOptimizer.Thread, atomic.LoadInt64(plan.writeOptimizer.Thread)-1)
	}
	return
}
func speedDel(event string, plan1 any) (res any) {
	plan := plan1.(*SchedulePlan2)
	atomic.SwapInt64(plan.ReadOptimizer.Thread, atomic.LoadInt64(plan.ReadOptimizer.Thread)+1)
	atomic.SwapInt64(plan.writeOptimizer.Thread, atomic.LoadInt64(plan.writeOptimizer.Thread)+1)
	return
}
func (plan *SchedulePlan2) speedResSend(event string, plan1 any) (res any) {
	if GlobalPConfigs.result.Teletypewriter == "bar" {
		if PlanContext.TaskBarSubsStatusGet(plan.subTaskInfo) != -2 {
			PlanContext.SpeedTaskBarAccumulate(plan1)
		}
	}
	return
}
func speedInit(plan *SchedulePlan2) mq.SpeedLimit {
	q := mq.NewSpeedMonitor(plan.Object.Schema, plan.Object.Table, plan.ReadOptimizer.ChunkSum)
	q.AvgRowsSize = plan.writeOptimizer.AvgRowsSize
	q.SpeedAddFunc = global.EFunc{FuncName: speedAdd, Params: plan}
	q.SpeedDelFunc = global.EFunc{FuncName: speedDel, Params: plan}
	q.SendResFunc = global.EFunc{FuncName: plan.speedResSend}
	return q
}

// 统一错误处理函数

func syncIsBaseQueryListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var plan *SchedulePlan2
	var wg sync.WaitGroup
	var event = "[syncIsBaseQueryListeners]"
	defer recoverException(event)
	defer func() {
		baseMsgSendBody.MsgCurryDone()
		baseMsgSendBody.MsgCurryWgDone()
	}()
	plan = message.Properties["plan"].(*SchedulePlan2)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": syncIsBaseQuery, "outFay": GlobalPConfigs.result.Teletypewriter})
	if syncGlobalInitActive(message) != nil {
		return mq.ReconsumeLater
	}
	plan.MQ.Wg, plan.MQ.Speed.Wg = &wg, &wg
	plan.MQ.AbnormalQuit, plan.MQ.NormalQuit = global.EFunc{FuncName: mqStatusMonitorAbnormalQuit, Params: plan}, global.EFunc{FuncName: mqStatusMonitorNormalQuit, Params: plan}
	plan.MQ.MqTableStatusMonitor(getTableName(plan.Object.Schema, plan.Object.Table))
	if syncBaseQuery(&wg, message) != nil {
		return mq.ReconsumeLater
	}
	wg.Wait()
	return mq.ConsumeSuccess
}
func taskObjectNew(logSeq int64, message mq.Message) (td Fp.TaskDistribution, err error) {
	var event = "[taskObjectNew]"
	defer func() {
		if re := recover(); re != nil {
			err = errors.New(fmt.Sprintf("%v defer recover An exception was captured. abnormal is %v ", event, re))
		}
		if err != nil {
			handleError(message, event, err)
		}
	}()
	segmentSql := message.Properties["segmentSql"].(parDef.IndexPart)
	plan := message.Properties["plan"].(*SchedulePlan2)
	if td, err = getScheduleTaskDistribution(plan, "left", fmt.Sprintf("%v", message.Properties["sourcePartitionName"]), time.Duration(GlobalPConfigs.rules.Sync.TimeOutNew.Segment)); err != nil {
		err = ref.ErrAddPrintf(event, err)
		log.ErrorLog().Error(fmt.Sprintf("get schedule task for table %v fail. error is {%v}", getTableName(plan.Object.Schema, plan.Object.Table), err))
		return
	}
	td.LogSeq = logSeq
	td.Second.Result = make(chan any, GlobalPConfigs.rules.QueueSize)
	td.IsConcurrentValue(segmentSql)
	return
}
func NewMonitorMsg(MsgModifyKey chan any, message mq.Message, msg map[string]any, successMsg, finishMsg []any) *mq.MonitorMsgMq {
	var plan = message.Properties["plan"].(*SchedulePlan2)
	q := mq.NewMonitorMsgMq()
	q.SendMsg = mq.ChanSendMsg{
		Table:       plan.Object.Table,
		MsgData:     plan.MQ.SendMsg.BaseMsg,
		MsgProducer: producer,
	}
	q.SpeedLimitSwitch = false
	q.SendMsg.MsgData = getMsgSendData(plan, msg)
	q.SendMsg.MsgSendFunc = SendMsgActionFunc(q.SendMsg.MsgData)
	q.SendMsg.MsgSuccessfulFunc = SendMsgExecSumFunc(successMsg)
	q.SendMsg.MsgFinishFunc = SendMsgExecSumFunc(finishMsg)
	q.SendMsg.MsgModifyKey = MsgModifyKey
	return q
}

// syncIsSecondQueryListeners IsSecondQuery 监听执行器
func syncIsSecondQueryListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var event = "[syncIsSecondQueryListeners]"
	var logSeq = rand.NewSource(time.Now().UnixNano()).Int63()
	var err error
	var td Fp.TaskDistribution
	defer recoverException(event)
	var plan = message.Properties["plan"].(*SchedulePlan2)
	var deleteMapInfo = map[string]any{"subTask": plan.subTaskInfo, "topic": syncIndexConcurrentQuery, "outFay": GlobalPConfigs.result.Teletypewriter}
	defer deleteTopic(deleteMapInfo)
	segmentSql := message.Properties["segmentSql"].(parDef.IndexPart)
	if abnormalErrorQuit(plan) || len(segmentSql.Error) > 0 { //Abnormal termination
		runErrorAction(message, errors.New(fmt.Sprintf("%v", segmentSql.Error)))
		return mq.ReconsumeLater
	}
	if td, err = taskObjectNew(logSeq, message); err != nil {
		handleError(message, event, err)
		return mq.ReconsumeLater
	}
	NewMonitorMsg(td.Second.Result, message, map[string]any{"logSeq": logSeq, "taskObject": message.Properties["taskObject"],
		"sendMsgKey": "sqlWhere", "topic": syncQueryTableSql,
		"sourcePartitionName": message.Properties["sourcePartitionName"]}, []any{[]any{plan.MQ.SecondMsg.ProductMsg}, int64(1)}, []any{[]any{plan.MQ.FirstMsg.CustomerMsg, plan.MQ.SecondMsg.ProductSendMsgFinishMark}, int64(1)}).ChanMonitorSendMsg("syncIndexConcurrentQueryListeners")
	return mq.ConsumeSuccess
}

// querySqlListeners QueryTableSql 监听执行器
func querySqlListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		plan     *SchedulePlan2
		tableSql string
		err      error
		event    = "[querySqlListeners]"
	)
	defer recoverException(event)
	plan = message.Properties["plan"].(*SchedulePlan2)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": syncQueryTableSql, "outFay": GlobalPConfigs.result.Teletypewriter})
	sqlWhere := message.Properties["sqlWhere"].(parDef.IndexPart)
	if abnormalErrorQuit(plan) || len(sqlWhere.Error) > 0 {
		return mq.ReconsumeLater
	}
	if tableSql, err = getQuerySql(message); err != nil || len(tableSql) == 0 {
		if err != nil {
			handleError(message, event, err)
		}
		plan.MQ.SecondMsgCustomerAdd()
		return mq.ConsumeSuccess
	}
	if !plan.MQ.SendMsgF(getMsgSendData(plan, map[string]any{"logSeq": sqlWhere.LogSeq, "taskObject": message.Properties["taskObject"],
		"strSql": tableSql, "topic": syncQueryTableData, "sourcePartitionName": message.Properties["sourcePartitionName"]})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}

// queryDataListeners queryData 监听执行器
func queryDataListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		stt   any
		err   error
		event = "[queryDataListeners]"
	)
	defer recoverException(event)
	var plan = message.Properties["plan"].(*SchedulePlan2)
	var deleteMapInfo = map[string]any{"subTask": plan.subTaskInfo, "topic": syncQueryTableData, "outFay": GlobalPConfigs.result.Teletypewriter}
	defer deleteTopic(deleteMapInfo)
	defer plan.MQ.Speed.ReadParallelDel()
	if abnormalErrorQuit(plan) {
		return mq.ReconsumeLater
	}
	plan.MQ.Speed.ReadParallelAdd()
	if stt, err = getRowsData(message); err != nil {
		handleError(message, event, err)
		return mq.ReconsumeLater
	}
	if barTerminalRowsUpdate(stt, "left", plan) != nil {
		return mq.ReconsumeLater
	}
	if !plan.MQ.SendMsgF(getMsgSendData(plan, map[string]any{"logSeq": message.Properties["logSeq"], "taskObject": message.Properties["taskObject"],
		"chunkData": stt, "topic": syncGeneralInsert, "sourcePartitionName": message.Properties["sourcePartitionName"]})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}

// generalInsertListeners generalInsert 监听执行器
func generalInsertListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var insertSql []any
	var err error
	var event = "[generalInsertListeners]"
	defer recoverException(event)
	var plan = message.Properties["plan"].(*SchedulePlan2)
	var deleteMapInfo = map[string]any{"subTask": plan.subTaskInfo, "topic": syncGeneralInsert, "outFay": GlobalPConfigs.result.Teletypewriter}
	defer deleteTopic(deleteMapInfo)
	if abnormalErrorQuit(plan) {
		return mq.ReconsumeLater
	}
	if insertSql, err = generalInsert(message); err != nil {
		handleError(message, event, err)
		return mq.ReconsumeLater
	}
	if !plan.MQ.SendMsgF(getMsgSendData(plan, map[string]any{"taskObject": message.Properties["taskObject"],
		"insertSql": insertSql, "topic": syncWriteInsert, "sourcePartitionName": message.Properties["sourcePartitionName"]})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess

}

// writeInsertListeners  监听执行器
func writeInsertListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var plan *SchedulePlan2
	defer func() {
		deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": syncWriteInsert, "outFay": GlobalPConfigs.result.Teletypewriter})
		plan.MQ.Speed.WriteParallelDel()
		plan.MQ.SecondMsgCustomerAdd()
	}()
	plan = message.Properties["plan"].(*SchedulePlan2)
	if abnormalErrorQuit(plan) {
		return mq.ReconsumeLater
	}
	plan.MQ.Speed.WriteParallelAdd()
	if writeInsert(message) != nil {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
